package org.shj.spark.sql

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.shj.spark.util.Util
import org.apache.spark.sql.SaveMode
import java.util.Date
import org.shj.spark.util.DateUtils
import org.apache.spark.sql.types.DateType

//Salary 不能设置为 Int, 否则会报错： Cannot up cast `salary` from bigint to int as it may truncate
case class Employee(deptCd: String, name: String, salary: Long, birthDt: String)

case class Person(name: String, age: Int)

/**
 * Dataset: 分布式的数据集合
 * DataFrame: 是面向列的Dataset (type DataFrame = Dataset[Row])
 */
object DatasetCreateMethod {
  def main(args: Array[String]){
    val spark = SparkSession.builder().appName("DatasetCreateMethod").master("local").getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    
    import spark.implicits._
    
    val filepath = Util.fullPath("employees.json")
    
    //方式一： 通过读json文件, 返回的是 DataFrame
    println("======= 方式一： 通过读json文件, 返回的是 DataFrame  =======")
    val df = spark.read.json(filepath)
    df.show()
    
    //方式二： 利用 case class 直接得到 Dataset[Person]
    /*println("======= 方式二： 利用 case class 直接得到 Dataset[Employee]  =======")
    val empDs = spark.read.json(filepath).as[Employee]
    empDs.show()*/
    
    //方式三： 把RDD转化成Dataset
    /*println("======= 方式三： 把RDD转化成Dataset, 使用 Bean  =======")
    val peopleDf = spark.sparkContext.textFile(Util.fullPath("people.txt"))
         .map(_.split(",\\s+"))
         .map(arr => Person(arr(0), arr(1).toInt))
         .toDF()
    peopleDf.show()*/
    
    
    //方式四： 通过编码的方式
    println("======= 方式四： 通过编码的方式, 不使用 Bean  =======")
    val peopleRDD = spark.sparkContext.textFile(Util.fullPath("people.txt"))
    val rowRdd = peopleRDD.map(line => {
      val arr = line.split(",\\s+")
      Row(arr(0), arr(1).toInt, arr(2))
    })
    
    val fields = Array(StructField("name", StringType, nullable=true),
                       StructField("age", IntegerType, nullable=true),
                       StructField("birth", StringType, nullable=true))
    val peopleDF = spark.createDataFrame(rowRdd, StructType(fields))
    peopleDF.foreach(row => {
      println(row.getAs[Date]("birth"))
    })
    //peopleDF.show()
    
    
    println("======= 方式五： 直接使用SQL  =======")
    //方式五： 直接使用SQL
    val sqlDF = spark.sql("select * from parquet.`D:/workspace/scala/sparkscala/file/users.parquet`")
    sqlDF.show()
    
    spark.stop()
    
  }
}